package com.niodata.dp.task.dao.handler;

import com.niodata.dp.mybatis.DpMybatisTypeHandler;
import com.niodata.dp.task.entity.JobParams;
import com.niodata.dp.util.JsonUtil;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.sql.CallableStatement;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import java.util.zip.ZipException;
import org.apache.hadoop.io.IOUtils;
import org.apache.ibatis.type.BaseTypeHandler;
import org.apache.ibatis.type.JdbcType;

public class JobParamsTypeHandler extends BaseTypeHandler<JobParams>
    implements DpMybatisTypeHandler {

  @Override
  public JdbcType getJdbcType() {
    return JdbcType.BLOB;
  }

  @Override
  public Class<JobParamsTypeHandler> getTypeHandlerClass() {
    return JobParamsTypeHandler.class;
  }

  @Override
  public Class<JobParams> getJavaClass() {
    return JobParams.class;
  }

  @Override
  public void setNonNullParameter(PreparedStatement ps, int i, JobParams parameter,
      JdbcType jdbcType) throws SQLException {
    try {
      byte[] bytes = encodeJobParams(parameter);
      ps.setBytes(i, bytes);
    } catch (IOException ioe) {
      throw new RuntimeException(ioe);
    }

  }

  @Override
  public JobParams getNullableResult(ResultSet rs, String columnName) throws SQLException {
    byte[] bytes = rs.getBytes(columnName);
    if (bytes == null || bytes.length == 0) {
      return null;
    }
    return decodeJobParams(bytes);
  }

  @Override
  public JobParams getNullableResult(ResultSet rs, int columnIndex) throws SQLException {
    byte[] bytes = rs.getBytes(columnIndex);
    if (bytes == null || bytes.length == 0) {
      return null;
    }
    return decodeJobParams(bytes);

  }

  @Override
  public JobParams getNullableResult(CallableStatement cs, int columnIndex) throws SQLException {
    byte[] bytes = cs.getBytes(columnIndex);
    if (bytes == null || bytes.length == 0) {
      return null;
    }
    return decodeJobParams(bytes);
  }

  /**
   * decode job params.
   *
   * @param bytes compressed or uncompressed bytes
   * @return jobparams
   */
  public JobParams decodeJobParams(byte[] bytes) {
    if (bytes == null) {
      return null;
    }
    try {
      GZIPInputStream zipInputStream = new GZIPInputStream(new ByteArrayInputStream(bytes));
      ByteArrayOutputStream baos = new ByteArrayOutputStream();
      IOUtils.copyBytes(zipInputStream, baos, 1024);
      String json = new String(baos.toByteArray());
      zipInputStream.close();
      return JsonUtil.fromJson(JobParams.class, json);
    } catch (ZipException e) {
      try {
        return JsonUtil.fromJson(JobParams.class, new String(bytes));
      } catch (IOException e1) {
        e1.printStackTrace();
        return null;
      }
    } catch (IOException ioe) {
      throw new RuntimeException("decode error");
    }
  }

  public byte[] encodeJobParams(JobParams jobParams) throws IOException {
    String json = JsonUtil.toJson(jobParams);
    ByteArrayOutputStream baos = new ByteArrayOutputStream();
    GZIPOutputStream zipOutputStream = new GZIPOutputStream(baos);
    zipOutputStream.write(json.getBytes());
    zipOutputStream.close();
    byte[] bytes = baos.toByteArray();
    return bytes;
  }
}
